import a_aa_amainpackage.a_o2odata_deal.config.config.{months, years}
import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.elasticsearch.spark._
/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2019/10/8 17:53
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object modify_cate {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.set("spark.debug.maxToStringFields", "500")
    conf.setMaster("local[*]")
    conf.set("es.nodes", "192.168.1.157")
    conf.set("es.port", "9200")
    conf.set("cluster.name", "O2OElastic")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.sql.caseSensitive", "true")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    sc.setLogLevel("WARN")
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")

    val value1: RDD[String] = sc.esJsonRDD(s"2020_tmall/tmall_2020_3",
      """
          {
     |  "query": {
     |    "bool": {
     |      "must": [
     |        {
     |          "term": {
     |            "firstCategoryId": {
     |              "value": "10021"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "secondCategoryId": {
     |              "value": "1002101"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "thirdCategoryId": {
     |              "value": "100210104"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "fourthCategoryId": {
     |              "value": "10021040399"
     |            }
     |          }
     |        }
     |      ]
     |    }
     |  }
     |}
    """.stripMargin).values
    ///////////////////////////
    val value2: RDD[String] = sc.esJsonRDD(s"2020_tmall/tmall_2020_3",
      """
          {
     |  "query": {
     |    "bool": {
     |      "must": [
     |        {
     |          "term": {
     |            "firstCategoryId": {
     |              "value": "10021"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "secondCategoryId": {
     |              "value": "1002101"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "thirdCategoryId": {
     |              "value": "100210104"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "fourthCategoryId": {
     |              "value": "10021019999"
     |            }
     |          }
     |        }
     |      ]
     |    }
     |  }
     |}
    """.stripMargin).values
    ////////////////////////////
    val value3: RDD[String] = sc.esJsonRDD(s"2020_tmall/tmall_2020_3",
      """
          {
     |  "query": {
     |    "bool": {
     |      "must": [
     |        {
     |          "term": {
     |            "firstCategoryId": {
     |              "value": "10021"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "secondCategoryId": {
     |              "value": "1002101"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "thirdCategoryId": {
     |              "value": "100210103"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "fourthCategoryId": {
     |              "value": "10021019999"
     |            }
     |          }
     |        }
     |      ]
     |    }
     |  }
     |}
    """.stripMargin).values
    ///////////////////////
    val value4: RDD[String] = sc.esJsonRDD(s"2020_tmall/tmall_2020_3",
      """
          {
     |  "query": {
     |    "bool": {
     |      "must": [
     |        {
     |          "term": {
     |            "firstCategoryId": {
     |              "value": "10021"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "secondCategoryId": {
     |              "value": "1002101"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "thirdCategoryId": {
     |              "value": "100210103"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "fourthCategoryId": {
     |              "value": "10099999999"
     |            }
     |          }
     |        }
     |      ]
     |    }
     |  }
     |}
    """.stripMargin).values
    /////////////////////////
    val value5: RDD[String] = sc.esJsonRDD(s"2020_tmall/tmall_2020_3",
      """
          {
     |  "query": {
     |    "bool": {
     |      "must": [
     |        {
     |          "term": {
     |            "firstCategoryId": {
     |              "value": "10021"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "secondCategoryId": {
     |              "value": "1002101"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "thirdCategoryId": {
     |              "value": "100210108"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "fourthCategoryId": {
     |              "value": "10099999999"
     |            }
     |          }
     |        }
     |      ]
     |    }
     |  }
     |}
    """.stripMargin).values
    //////////////////////
    val value6: RDD[String] = sc.esJsonRDD(s"2020_tmall/tmall_2020_3",
      """
          {
     |  "query": {
     |    "bool": {
     |      "must": [
     |        {
     |          "term": {
     |            "firstCategoryId": {
     |              "value": "10013"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "secondCategoryId": {
     |              "value": "1001506"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "thirdCategoryId": {
     |              "value": "100150604"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "fourthCategoryId": {
     |              "value": "10015060499"
     |            }
     |          }
     |        }
     |      ]
     |    }
     |  }
     |}
    """.stripMargin).values
    ///////////////////////////
    val value7: RDD[String] = sc.esJsonRDD(s"2020_tmall/tmall_2020_3",
      """
          {
     |  "query": {
     |    "bool": {
     |      "must": [
     |        {
     |          "term": {
     |            "firstCategoryId": {
     |              "value": "10025"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "secondCategoryId": {
     |              "value": "1002508"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "thirdCategoryId": {
     |              "value": "100250802"
     |            }
     |          }
     |        },{
     |          "term": {
     |            "fourthCategoryId": {
     |              "value": "10025089999"
     |            }
     |          }
     |        }
     |      ]
     |    }
     |  }
     |}
    """.stripMargin).values

    val value=value1.union(value2).union(value3).union(value4).union(value5).union(value6).union(value7)

     //val frame: DataFrame = sqlContext.read.json(value1)
     val frame1: DataFrame = sqlContext.read.json(value)
    frame1.write.json(s"s3a://o2o-dataproces-group/panzonghao/tmall/${years}/${months}/cate_backs")
    frame1.createOrReplaceTempView("frame1")

    val frame = sqlContext.sql(
      s"""
         |select
         |*,
         |case
         |when firstCategoryId='10021' and secondCategoryId='1002101' and thirdCategoryId='100210104' and fourthCategoryId='10021040399' then '10021010499'
         |when firstCategoryId='10021' and secondCategoryId='1002101' and thirdCategoryId='100210104' and fourthCategoryId='10021019999' then '10021010499'
         |when firstCategoryId='10021' and secondCategoryId='1002101' and thirdCategoryId='100210103' and fourthCategoryId='10021019999' then '10021010399'
         |when firstCategoryId='10021' and secondCategoryId='1002101' and thirdCategoryId='100210103' and fourthCategoryId='10099999999' then '10021010399'
         |when firstCategoryId='10021' and secondCategoryId='1002101' and thirdCategoryId='100210108' and fourthCategoryId='10099999999' then '10021010899'
         |when firstCategoryId='10025' and secondCategoryId='1002508' and thirdCategoryId='100250802' and fourthCategoryId='10025089999' then '10025080299'
         |else fourthCategoryId end as fourth,
         |
         |case
         |when firstCategoryId='10013' and secondCategoryId='1001506' and thirdCategoryId='100150604' and fourthCategoryId='10015060499' then '10015'
         |else firstCategoryId end as first
         |from frame1
       """.stripMargin).drop("fourthCategoryId","firstCategoryId").withColumnRenamed("fourth","fourthCategoryId").withColumnRenamed("first","firstCategoryId")



    val result = frame.toJSON.rdd.map(line => {
      val lines = JSON.parseObject(line)
      val evaluates = lines.getOrDefault("evaluates", "-1").toString
      val Base_Info = lines.getOrDefault("Base_Info", "-1").toString
      var ev = "-1"
      var str = new JSONObject
      if (evaluates.equals("-1")) {
        ev = "{\"fuyi\":\"-1\"}"
      } else {
        str = JSON.parseObject(evaluates)
      }
      if (!ev.contains("fuyi")) {
        lines.put("evaluates", str)
      } else {
        val evs = JSON.parseObject(ev)
        lines.put("evaluates", evs)
      }
      lines.put("Base_Info",Base_Info)
      lines
    })
    result.saveToEs(s"2020_tmall/tmall_2020_3", Map("es.mapping.id" -> "good_id"))
    sc.stop()
  }
}

















